import { StringIndexer, BackendServices, isUndefined, camelCaseToUnderscore } from "mmb";
import * as nsq from "nsqjs";
import { getLookupdAddress, getNsqdAddress } from "../store/helper";

export interface NsqContext {
  subsMap: StringIndexer;
  subCommand: nsq.Reader | undefined;
  pubAvailable: boolean;
  pubRetryMax: number;
  pubRetry: number;
  pubRetryInterval: number;
  msgSignal: string;
  topicOfCommand: string;
  channelOfCommand: BackendServices;
  pub: nsq.Writer | undefined;
}

const ctx: NsqContext = {
  subsMap: {},
  subCommand: undefined,
  pubAvailable: false,
  pubRetryMax: 12 * 30, // 重试半小时，每次间隔5秒
  pubRetry: 0,
  pubRetryInterval: 5000,
  msgSignal: "^V^",
  topicOfCommand: `${BackendServices.BGAPI}_command`,
  channelOfCommand: BackendServices.BGAPI,
  pub: undefined,
};

export const isPubAvailable = () => {
  return !isUndefined(ctx.pubAvailable);
};

const makeDeferPublishForTesing = () => {
  if (isUndefined(ctx.pub)) {
    console.log("pub not available");
    return;
  }
  ctx.pub.deferPublish(ctx.topicOfCommand, [ctx.msgSignal], 2000, (err) => {
    if (err) {
      return console.error(err.message);
    }
    console.log(`Message ${ctx.msgSignal} sent successfully`);
    ctx.pubAvailable = true;
  });
};

export const makeSub = (topic: string, channel: string) => {
  return new nsq.Reader(topic, channel, {
    lookupdHTTPAddresses: getLookupdAddress()!,
  });
};

export const getPub = () => {
  if (!ctx.pubAvailable) {
    throw Error("pub is not available");
  }
  return ctx.pub!;
};

export const makeTopicOfPub = (topic: string) => {
  return `${BackendServices.BGAPI}-${topic}`;
};

export const wrapToMessage = (o: StringIndexer) => {
  return { ts: Date.now(), data: camelCaseToUnderscore(o) };
};

export const cmd = (commandStr: string) => {
  if (!ctx.pubAvailable) {
    console.log("pub not available");
    return "err:pub not available";
  }
  ctx.pub!.publish(ctx.topicOfCommand, commandStr, (err) => {
    if (err) {
      return console.error(err.message);
    }
    console.log(`cmd ${commandStr} published`);
  });
  return "ok";
};

export const init = () => {
  if (isUndefined(ctx.subCommand)) {
    ctx.subCommand = makeSub(ctx.topicOfCommand, ctx.channelOfCommand);
  }
  ctx.subCommand.connect();
  ctx.subCommand
    .on("ready", () => {
      console.log(`sub ${ctx.topicOfCommand} ready ...`);
    })
    .on("error", (err) => {
      console.log(`sub ${ctx.topicOfCommand}  error => `, err);
    })
    .on("nsqd_connected", () => {
      console.log("subCommand connected ");
    })
    .on("nsqd_closed", () => {
      console.log("subCommand closed => ");
    })
    .on("message", (msg: nsq.Message) => {
      console.log("Received message [%s]", msg.id);
      let msgBody = msg.body.toString();
      console.log("message body", msg.body.toString());
      if (msg.hasResponded) {
        console.log(`消息[${msg.id}]消息已被响应`);
        msg.finish();
        return;
      }
      switch (msgBody) {
        case ctx.msgSignal:
          console.log("received signal");
          break;
        case `ping`:
          console.log("received ping");
          break;
        default:
          break;
      }
      msg.finish();
    });

  if (isUndefined(ctx.pub)) {
    const [pubHost, pubPort] = getNsqdAddress()!.split(":");
    ctx.pub = new nsq.Writer(pubHost, Number(pubPort));
  }
  ctx.pub.connect();
  ctx.pub
    .on("ready", () => {
      console.log("pub ready ===>");
      makeDeferPublishForTesing();
    })
    .on("error", (err) => {
      console.log("pub error => ", err);
    })
    .on("closed", () => {
      console.log("pub closed");
      if (ctx.pubRetry < ctx.pubRetryMax) {
        setTimeout(() => {
          ctx.pubRetry += 1;
          ctx.pub!.connect();
          console.log(`====== pub retry connect on ${ctx.pubRetry} ======`);
        }, ctx.pubRetryInterval);
      }
    });
};
